package com.merge.task;

import com.common.ArrayUtils;
import com.merge.FileRead;
import com.merge.constant.FileConstant;
import com.merge.enums.TaskStatus;
import com.merge.pojo.Data;
import com.merge.pojo.SliceFile;
import com.merge.queue.SliceQueue;
import com.merge.queue.SliceQueueManger;

/**
 * 监控分片队列任务类
 */
public class MonitorSliceQueueTask extends Task {

    public static final int TOP_NUM = 5000;

    private Thread mergeTaskThread = new Thread(new MergeTask(1));

    public MonitorSliceQueueTask(int id) {
        super(id, "monitor_slice_queue_task");
    }

    @Override
    public void run() {
        SliceQueue<Data>[] queues = SliceQueueManger.sliceQueueMapToArr();
        if (ArrayUtils.empty(queues)) setStatus(TaskStatus.STOP.getValue());
        int status = getStatus();
        System.out.println("开启监控slice_queues任务");
        int finishFileNum = 0;
        int index = 1;
        //task的status是run状态，一直监控
        while (status == TaskStatus.RUN.getValue()) {
            try {
                Thread.sleep(0);
                //某个队列中的size小于TOP_NUM,从对应分片文件中offer值
                for (SliceQueue<Data> queue : queues) {
                    if (queue.montitor == 0) continue;
                    if (queue.getSize() < TOP_NUM) {
                        int count = TOP_NUM - queue.getSize();
                        while (count-- > 0) {
                            Data data = FileRead.getNextLineData(queue.getSliceFile(), queue.getName());
                            if (data == null) {
                                //对应的分片文件数据已读完，可以关闭该队列监控
                                System.out.println("对应的分片文件:"+ queue.getSliceFile().getName()+ "数据已读完，可以关闭该队列监控");
                                finishFileNum++;
                                if (finishFileNum == queues.length) {
                                    System.out.println("所有分片文件数据已读完，关闭监控程序");
                                    status = TaskStatus.STOP.getValue();
                                    break;
                                }
                                queue.montitor = 0;
                                break;
                            }
                            queue.offer(data);
                        }
                    }
                }
                //当第一次开启监控任务时，才能启动合并任务
                if (index++ == 1) {
                    System.out.println("开始启动合并任务");
                    mergeTaskThread.start();
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }


    public static void main(String[] args) {
        int num = 4;
        int index = 1;
        for (int i = 0; i < num; i++) {
            String name = FileConstant.SLICE_FILE_PROFIX + (i + 1);
            SliceFile sliceFile = new SliceFile(name, 10);
            SliceQueueManger.addSliceQueueMap(sliceFile);
        }
        MonitorSliceQueueTask monitorSliceQueueTask = new MonitorSliceQueueTask(1);
        Thread t1 = new Thread(monitorSliceQueueTask);
        t1.start();
    }
}
